package com.atguigu.gmall.app.dwd.log;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.util.DateFormatUtils;
import com.atguigu.gmall.util.MyKafkaUtils;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;


/**
 * 数据流：web/app -> Nginx -> 日志服务器(.log) -> Flume -> Kafka(ODS) -> FlinkApp -> Kafka(DWD)
 * 程  序：     Mock(lg.sh) -> Flume(f1) -> Kafka(ZK) -> BaseLogApp -> Kafka(ZK)
 *
 * @author : ranzlupup
 * @since : 2023/6/2 09:49
 */
public class BaseLogApp {
    public static void main(String[] args) throws Exception {
        //! 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1); //生产环境中设置为Kafka主题的分区数

        //1.1 开启CheckPoint
        //env.enableCheckpointing(5 * 60000L, CheckpointingMode.EXACTLY_ONCE);
        //env.getCheckpointConfig().setCheckpointTimeout(10 * 60000L);
        //env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        //env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5000L));

        //1.2 设置状态后端
        //env.setStateBackend(new HashMapStateBackend());
        //env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/211126/ck");
        //System.setProperty("HADOOP_USER_NAME", "atguigu");

        //! 2.消费Kafka FLINK_ODS_LOG 主题的数据创建流(base log)
        String topicName = "FLINK_ODS_LOG";
        String groupId = "FLINK_ODS_LOG_GROUP";
        FlinkKafkaConsumer<String> flinkKafkaConsumer = MyKafkaUtils.getFlinkKafkaConsumer(topicName, groupId);

        DataStreamSource<String> kafkaDSSource = env.addSource(flinkKafkaConsumer);

        //! 3.过滤掉非JSON格式的数据 & 将每行数据转换为JSON对象
        OutputTag<String> dirtyOutputTag = new OutputTag<String>("Dirty") {

        };
        SingleOutputStreamOperator<JSONObject> jsonObjectDS = kafkaDSSource.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String value, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                        try {
                            JSONObject jsonObject = JSONObject.parseObject(value);
                            collector.collect(jsonObject);
                        } catch (Exception e) {
                            // 如果是脏数据，写入侧输出流
                            context.output(dirtyOutputTag, value);
                        }
                    }
                }
        );
        // 获取侧输出流脏数据并打印
        DataStream<String> dirtyDS = jsonObjectDS.getSideOutput(dirtyOutputTag);
        // dirtyDS.print("Dirty>>>>>>>>>>>>>>>>>>");
        dirtyDS.print("Dirty>>>>>>>>>>>>>>>>>>");

        //! 4.按照mid分组
        KeyedStream<JSONObject, String> keyedDStream = jsonObjectDS.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject value) throws Exception {
                return value.getJSONObject("common").getString("mid");
            }
        });

        //! 5.使用状态编程做新老访客标记校验
        SingleOutputStreamOperator<JSONObject> jsonObjWithNewFlagDStream = keyedDStream.map(new RichMapFunction<JSONObject, JSONObject>() {
            private ValueState<String> lastVisitState;

            @Override
            public void open(Configuration parameters) throws Exception {
                lastVisitState = getRuntimeContext().getState(new ValueStateDescriptor<String>("last-visit-state", String.class));
            }

            @Override
            public JSONObject map(JSONObject value) throws Exception {
                // 获取is_new标记 & ts 并将时间戳转换为年月日
                String isNew = value.getJSONObject("common").getString("is_new");
                Long ts = value.getLong("ts");
                String curDate = DateFormatUtils.toDate(ts);

                // 获取状态中的日期
                String lastDate = lastVisitState.value();

                // 判断is_new标记是否为 "1"，"1"为新用户
                if ("1".equals(isNew)) {
                    // 如果is_new为 "1"
                    if (lastDate == null) {
                        // 如果状态中不存在日期，证明该用户是新用户，将数据中的日期存入状态
                        lastVisitState.update(curDate);
                    } else if (!lastDate.equals(curDate)) {
                        // 如果不为null，证明不是新用户，且数据日期与状态日期不相等，证明以前存入过状态
                        // 则将is_new该为 "0"
                        value.getJSONObject("common").put("is_new", 0);
                    }
                } else {
                    // 如果is_new为 "0"，说明不是新用户，状态中按道理讲应该存有日期
                    if (lastDate == null) {
                        // 如果状态中没有日期，可能是用户在程序执行前就已经不是新用户了
                        // 将昨天的日期存入状态
                        lastVisitState.update(DateFormatUtils.toDate(ts - 24 * 60 * 60 * 1000L));
                    }
                }
                return value;
            }
        });

        //! 6.使用侧输出流进行分流处理
        OutputTag<String> startTag = new OutputTag<String>("start") {

        };
        OutputTag<String> displayTag = new OutputTag<String>("display") {

        };
        OutputTag<String> actionTag = new OutputTag<String>("action") {

        };
        OutputTag<String> errorTag = new OutputTag<String>("error") {

        };
        SingleOutputStreamOperator<String> pageDS = jsonObjWithNewFlagDStream.process(
                new ProcessFunction<JSONObject, String>() {

                    @Override
                    public void processElement(JSONObject value, ProcessFunction<JSONObject, String>.Context ctx, Collector<String> out) throws Exception {
                        // 尝试获取错误信息
                        String err = value.getString("err");
                        if (err != null) {
                            // 如果error不为null，将error数据写入到error侧输出流
                            // System.out.println("err======>" + err);
                            ctx.output(errorTag, value.toJSONString());
                        }
                        // 移除错误信息
                        value.remove("err");

                        // 尝试获取启动信息
                        String start = value.getString("start");
                        if (start != null) {
                            // 将数据写到start侧输出流
                            ctx.output(startTag, start);
                        } else {
                            // start为null
                            // 获取公共信息common、页面id - page_id、时间戳ts
                            String common = value.getString("common");
                            String pageId = value.getJSONObject("page").getString("page_id");
                            Long ts = value.getLong("ts");

                            // 尝试获取曝光数据
                            JSONArray displays = value.getJSONArray("displays");
                            if (displays != null && displays.size() > 0) {
                                // 遍历曝光数据，写到display侧输出流
                                for (int i = 0; i < displays.size(); i++) {
                                    JSONObject display = displays.getJSONObject(i);
                                    display.put("common", common);
                                    display.put("page_id", pageId);
                                    display.put("ts", ts);
                                    ctx.output(displayTag, display.toJSONString());
                                }
                            }

                            // 尝试获取动作数据
                            JSONArray actions = value.getJSONArray("actions");
                            if (actions != null && actions.size() > 0) {
                                //遍历曝光数据&写到display侧输出流
                                for (int i = 0; i < actions.size(); i++) {
                                    JSONObject action = actions.getJSONObject(i);
                                    action.put("common", common);
                                    action.put("page_id", pageId);
                                    ctx.output(actionTag, action.toJSONString());
                                }
                            }

                            // 移除曝光数据和动作数据，并且将页面日志写入到主流
                            value.remove("displays");
                            value.remove("actions");
                            out.collect(value.toJSONString());
                        }
                    }
                }
        );

        //! 7.提取各个侧输出流数据
        DataStream<String> startDS = pageDS.getSideOutput(startTag);
        DataStream<String> displayDS = pageDS.getSideOutput(displayTag);
        DataStream<String> actionDS = pageDS.getSideOutput(actionTag);
        DataStream<String> errorDS = pageDS.getSideOutput(errorTag);


        //! 8.将数据打印并写入对应的主题
        pageDS.print("Page>>>>>>>>>>");
        startDS.print("Start>>>>>>>>");
        displayDS.print("Display>>>>");
        actionDS.print("Action>>>>>>");
        errorDS.print("Error>>>>>>>>");

        String page_topic = "FLINK_DWD_PAGE_LOG";
        String start_topic = "FLINK_DWD_START_LOG";
        String display_topic = "FLINK_DWD_DISPLAY_LOG";
        String action_topic = "FLINK_DWD_ACTION_LOG";
        String error_topic = "FLINK_DWD_ERROR_LOG";

        pageDS.addSink(MyKafkaUtils.getFlinkKafkaProducer(page_topic));
        startDS.addSink(MyKafkaUtils.getFlinkKafkaProducer(start_topic));
        displayDS.addSink(MyKafkaUtils.getFlinkKafkaProducer(display_topic));
        actionDS.addSink(MyKafkaUtils.getFlinkKafkaProducer(action_topic));
        errorDS.addSink(MyKafkaUtils.getFlinkKafkaProducer(error_topic));

        //! 9.启动任务
        env.execute("BaseLogApp");
    }
}
